package com.hive;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.Seekable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.*;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.LineRecordReader;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapreduce.lib.input.CompressedSplitLineReader;
import org.apache.hadoop.mapreduce.lib.input.SplitLineReader;
import org.apache.hadoop.mapreduce.lib.input.UncompressedSplitLineReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.InputStream;
public class UserRecordReader implements RecordReader<LongWritable,Text> {
    private static final Logger LOG =
            LoggerFactory.getLogger(LineRecordReader.class.getName());
    private CompressionCodecFactory compressionCodecs;
    private long start;
    private long pos;
    private long end;
    private SplitLineReader in;
    private FSDataInputStream fileIn;
    private final Seekable filePosition;
    int maxLineLength;
    private CompressionCodec codec;
    private Decompressor decompressor;
    public UserRecordReader(Configuration job, FileSplit split) throws
            IOException {
        this(job, split, (byte[])null);
    }
    public UserRecordReader(Configuration job, FileSplit split, byte[]
            recordDelimiter) throws IOException {
        this.compressionCodecs = null;
        this.maxLineLength =
                job.getInt("mapreduce.input.linerecordreader.line.maxlength", 2147483647);
        this.start = split.getStart();
        this.end = this.start + split.getLength();
        Path file = split.getPath();
        this.compressionCodecs = new CompressionCodecFactory(job);
        this.codec = this.compressionCodecs.getCodec(file);
        FileSystem fs = file.getFileSystem(job);
        this.fileIn = fs.open(file);
        if (this.isCompressedInput()) {
            this.decompressor = CodecPool.getDecompressor(this.codec);
            if (this.codec instanceof SplittableCompressionCodec) {
                SplitCompressionInputStream cIn =
                        ((SplittableCompressionCodec)this.codec).createInputStream(this.fileIn,
                                this.decompressor, this.start, this.end,
                                SplittableCompressionCodec.READ_MODE.BYBLOCK);
                this.in = new CompressedSplitLineReader(cIn, job,
                        recordDelimiter);
                this.start = cIn.getAdjustedStart();
                this.end = cIn.getAdjustedEnd();
                this.filePosition = cIn;
            } else {
                if (this.start != 0L) {
                    throw new IOException("Cannot seek in " +
                            this.codec.getClass().getSimpleName() + " compressed stream");
                }
                this.in = new
                        SplitLineReader(this.codec.createInputStream(this.fileIn, this.decompressor),
                        job, recordDelimiter);
                this.filePosition = this.fileIn;
            }
        } else {
            this.fileIn.seek(this.start);
            this.in = new UncompressedSplitLineReader(this.fileIn, job,
                    recordDelimiter, split.getLength());
            this.filePosition = this.fileIn;
        }
        if (this.start != 0L) {
            this.start += (long)this.in.readLine(new Text(), 0,
                    this.maxBytesToConsume(this.start));
        }
        this.pos = this.start;
    }
    public UserRecordReader(InputStream in, long offset, long endOffset, int
            maxLineLength) {

        this(in, offset, endOffset, maxLineLength, (byte[])null);
    }
    public UserRecordReader(InputStream in, long offset, long endOffset, int
            maxLineLength, byte[] recordDelimiter) {
        this.compressionCodecs = null;
        this.maxLineLength = maxLineLength;
        this.in = new SplitLineReader(in, recordDelimiter);
        this.start = offset;
        this.pos = offset;
        this.end = endOffset;
        this.filePosition = null;
    }
    public UserRecordReader(InputStream in, long offset, long endOffset,
                            Configuration job) throws IOException {
        this(in, offset, endOffset, job, (byte[])null);
    }
    public UserRecordReader(InputStream in, long offset, long endOffset,
                            Configuration job, byte[] recordDelimiter) throws IOException {
        this.compressionCodecs = null;
        this.maxLineLength =
                job.getInt("mapreduce.input.linerecordreader.line.maxlength", 2147483647);
        this.in = new SplitLineReader(in, job, recordDelimiter);
        this.start = offset;
        this.pos = offset;
        this.end = endOffset;
        this.filePosition = null;
    }
    public LongWritable createKey() {
        return new LongWritable();
    }
    public Text createValue() {
        return new Text();
    }
    private boolean isCompressedInput() {
        return this.codec != null;
    }
    private int maxBytesToConsume(long pos) {
        return this.isCompressedInput() ? 2147483647 :
                (int)Math.max(Math.min(2147483647L, this.end - pos), (long)this.maxLineLength);
    }
    private long getFilePosition() throws IOException {
        long retVal;
        if (this.isCompressedInput() && null != this.filePosition) {
            retVal = this.filePosition.getPos();
        } else {
            retVal = this.pos;
        }
        return retVal;
    }
    private int skipUtfByteOrderMark(Text value) throws IOException {
        int newMaxLineLength = (int)Math.min(3L + (long)this.maxLineLength,
                2147483647L);
        int newSize = this.in.readLine(value, newMaxLineLength,
                this.maxBytesToConsume(this.pos));
        this.pos += (long)newSize;
        int textLength = value.getLength();
        byte[] textBytes = value.getBytes();
        if (textLength >= 3 && textBytes[0] == -17 && textBytes[1] == -69 &&
                textBytes[2] == -65) {
            LOG.info("Found UTF-8 BOM and skipped it");
            textLength -= 3;
            newSize -= 3;
            if (textLength > 0) {
                textBytes = value.copyBytes();
                value.set(textBytes, 3, textLength);
            } else {
                value.clear();
            }
        }
        return newSize;
    }
    public synchronized boolean next(LongWritable key, Text value) throws
            IOException {
        while(this.getFilePosition() <= this.end ||
                this.in.needAdditionalRecordAfterSplit()) {
            key.set(this.pos);
            int newSize;
            newSize = this.in.readLine(value, this.maxLineLength,
                    this.maxBytesToConsume(this.pos));
            String str=value.toString().replaceAll("\\|\\|","\\|");
            value.set(str);
            this.pos += (long)newSize;
            if (newSize == 0) {
                return false;
            }
            if (newSize < this.maxLineLength) {
                return true;
            }
            LOG.info("Skipped line of size " + newSize + " at pos " + (this.pos
                    - (long)newSize));
        }
        return false;
    }
    public synchronized float getProgress() throws IOException {
        return this.start == this.end ? 0.0F : Math.min(1.0F, (float)
                (this.getFilePosition() - this.start) / (float)(this.end - this.start));
    }
    public synchronized long getPos() throws IOException {
        return this.pos;
    }
    public synchronized void close() throws IOException {
        try {
            if (this.in != null) {
                this.in.close();
            }
        } finally {
            if (this.decompressor != null) {
                CodecPool.returnDecompressor(this.decompressor);
                this.decompressor = null;
            }
        }
    }
    /** @deprecated */
    @Deprecated
    public static class LineReader extends org.apache.hadoop.util.LineReader {
        LineReader(InputStream in) {
            super(in);
        }
        LineReader(InputStream in, int bufferSize) {
            super(in, bufferSize);
        }
        public LineReader(InputStream in, Configuration conf) throws IOException
        {
            super(in, conf);
        }
        LineReader(InputStream in, byte[] recordDelimiter) {
            super(in, recordDelimiter);
        }
        LineReader(InputStream in, int bufferSize, byte[] recordDelimiter) {
            super(in, bufferSize, recordDelimiter);
        }
        public LineReader(InputStream in, Configuration conf, byte[]
                recordDelimiter) throws IOException {
            super(in, conf, recordDelimiter);
        }
    }
}